package com.msb;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WcMainCluser {
    public static void main(String[] args) throws Exception {
        // 1 获取配置信息，或者job对象实例
        Configuration configuration = new Configuration();

        /*本地提交集群运行
        * */
       // configuration.set("mapreduce.app-submission.cross-platform", "true");
        configuration.set("fs.defaultFS","hdfs://node01:9000");

        Job job = Job.getInstance(configuration);

        // 6 指定本程序的jar包所在的本地路径
//        job.setJar("/home/admin/wc.jar");
        job.setJarByClass(WcMainCluser.class);

        // 2 指定本业务job要使用的mapper/Reducer业务类
        job.setMapperClass(WcMapper.class);
        job.setReducerClass(WcReduce.class);

        // 3 指定mapper输出数据的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 4 指定最终输出的数据的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 5 指定job的输入原始文件所在目录
        FileInputFormat.setInputPaths(job, new Path( "hdfs://node01:9000/input/text"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://node01:9000/output"));

        // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
        job.submit();
        boolean result = job.waitForCompletion(true);
        System.exit(result?0:1);
    }
}
